package cn.itcast.flink.batch;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * @author lilulu
 * @date 2023-03-28 23:56
 */
public class BatchWordCount {
    /**
     * 1、执行环境
     * 2、数据源source
     * 3、数据转换transformation
     * 4、数据接收器sink
     * 5、触发执行execute
     */
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSource<String> inputDataSet = env.readTextFile("datas/words.txt");
        /**
         * spark hadoop hive
         * 1、分割单词：
         *      flapMap:spark,hadoop,hive
         * 2、转换二元组：
         *      map:(spark,1),(hadoop,1),(hive,1)
         * 3、分组：
         *      groupBy(0):spark:[(spark,1)] hadoop:[(hadoop,1)],hive:[(hive:1)]
         * 4、聚合统计求和
         *      sum(1): spark:1,spark:2,hive:3
         */
        //分割单词
        FlatMapOperator<String, String> wordDateSet = inputDataSet.flatMap(
                new FlatMapFunction<String, String>() {
                    public void flatMap(String line, Collector<String> collector) throws Exception {
                        String[] words = line.trim().split("\\s+");
                        for (String word : words) {
                            collector.collect(word);
                        }
                    }
                }
        );
        //二元组
        MapOperator<String, Tuple2<String, Integer>> tupleDataSet = wordDateSet.map(
                new MapFunction<String, Tuple2<String, Integer>>() {
                    public Tuple2<String, Integer> map(String word) throws Exception {
                        return Tuple2.of(word, 1);
                    }
                }
        );
        AggregateOperator<Tuple2<String, Integer>> resultDataSet = tupleDataSet.groupBy(0).sum(1);
        resultDataSet.print();
    }
}
